insert overwrite table jms_dm.dm_bagging_overdue_summary_dt partition(dt='{{ execution_date | cst_ds }}')
select
    to_date(input_time) as input_time
     ,scan_site_name      as scan_site_name
     ,scan_site_code      as scan_site_code
     ,bagg_cnt            as bagg_cnt
     ,bagg_timely_cnt     as bagg_timely_cnt
     ,bagg_not_timely_cnt as bagg_not_timely_cnt
     ,(bagg_not_timely_cnt / bagg_cnt * 100) as bagg_timely_rate
from(
    select
        to_date(bagg.input_time)        as input_time     --上传时间
         ,bagg.scan_site                  as scan_site_name --扫描网点
         ,bagg.scan_site_code             as scan_site_code --扫描网点编号
         ,count(bagg.waybill_no) as bagg_cnt       --建包扫描操作量
         ,sum(case when (bagg.input_time > send.scan_time and send.scan_time > bagg.scan_time) or send.waybill_no is null  or
                        (bagg.scan_time  > send.scan_time and (unix_timestamp(bagg.input_time) - unix_timestamp(bagg.scan_time)) / 60 > 40) then 1 else 0 end) as bagg_not_timely_cnt
         ,sum(case when  (unix_timestamp(bagg.input_time) - unix_timestamp(bagg.scan_time)) / 60 <= 40 then 1 else 0 end) as bagg_timely_cnt
    from (
        select *
        from(
            select
                waybill_no,input_time,scan_site,scan_site_code,scan_time,
                row_number() over(partition by waybill_no,scan_site_code order by scan_time desc) as rn
            from jms_dwd.dwd_tab_barscan_bagging_base_dt
            where dt = date_sub('{{ execution_date | cst_ds }}',1)
              and waybill_no not like '%B%'
        ) a where a.rn = 1
    ) bagg
    left join(
        select *
        from(
            select *,row_number() over(partition by waybill_no,scan_site_code order by scan_time desc) as rn  --去重
            from(
                select
                    waybill_no
                     ,scan_time
                     ,next_station
                     ,next_station_code
                     ,scan_site_code
                from jms_dwd.dwd_tab_barscan_send_base_dt
                where dt between date_sub('{{ execution_date | cst_ds }}',1) and  date_add('{{ execution_date | cst_ds }}',1)
                union all
                select
                    billcode         as waybill_no
                     ,scantime         as scan_time
                     ,nextstation      as next_station
                     ,nextstationcode  as next_station_code
                     ,scansitecode     as scan_site_code
                from jms_ods.tab_barscan_centersend
                where dt between date_sub('{{ execution_date | cst_ds }}',1) and  date_add('{{ execution_date | cst_ds }}',1)
            ) a
        ) a where a.rn = 1
    )send on bagg.waybill_no = send.waybill_no
        and bagg.scan_site_code = send.scan_site_code  --网点
    group by to_date(bagg.input_time)
           ,bagg.scan_site
           ,bagg.scan_site_code
) a
distribute by dt,pmod(hash(rand()),1);